// Copyright (c) 2025 coze-dev Authors
// SPDX-License-Identifier: Apache-2.0

package service

import (
	"context"
	"fmt"
	"strconv"
	"sync"
	"time"

	"github.com/coze-dev/coze-loop/backend/infra/redis"
	tconv "github.com/coze-dev/coze-loop/backend/modules/observability/application/convertor/task"
	taskrepo "github.com/coze-dev/coze-loop/backend/modules/observability/domain/task/repo"
	"golang.org/x/sync/errgroup"

	"github.com/bytedance/gg/gptr"
	"github.com/coze-dev/coze-loop/backend/infra/middleware/session"
	"github.com/coze-dev/coze-loop/backend/kitex_gen/coze/loop/observability/domain/annotation"
	"github.com/coze-dev/coze-loop/backend/kitex_gen/coze/loop/observability/domain/common"
	"github.com/coze-dev/coze-loop/backend/kitex_gen/coze/loop/observability/domain/dataset"
	"github.com/coze-dev/coze-loop/backend/kitex_gen/coze/loop/observability/trace"
	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/component/config"
	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/component/metrics"
	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/component/mq"
	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/component/rpc"
	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/component/tenant"
	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/trace/entity"
	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/trace/entity/loop_span"
	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/trace/repo"
	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/trace/service/trace/span_filter"
	"github.com/coze-dev/coze-loop/backend/modules/observability/domain/trace/service/trace/span_processor"
	obErrorx "github.com/coze-dev/coze-loop/backend/modules/observability/pkg/errno"
	"github.com/coze-dev/coze-loop/backend/pkg/errorx"
	"github.com/coze-dev/coze-loop/backend/pkg/json"
	"github.com/coze-dev/coze-loop/backend/pkg/lang/goroutine"
	"github.com/coze-dev/coze-loop/backend/pkg/lang/ptr"
	"github.com/coze-dev/coze-loop/backend/pkg/logs"
	timeutil "github.com/coze-dev/coze-loop/backend/pkg/time"
	"github.com/samber/lo"
)

type ListSpansReq struct {
	WorkspaceID           int64
	ThirdPartyWorkspaceID string
	StartTime             int64 // ms
	EndTime               int64 // ms
	Filters               *loop_span.FilterFields
	Limit                 int32
	DescByStartTime       bool
	PageToken             string
	PlatformType          loop_span.PlatformType
	SpanListType          loop_span.SpanListType
}

type ListSpansResp struct {
	Spans         loop_span.SpanList
	NextPageToken string
	HasMore       bool
}

type ListPreSpanReq struct {
	WorkspaceID        int64
	StartTime          int64 // ms
	TraceID            string
	SpanID             string
	PreviousResponseID string
	PlatformType       loop_span.PlatformType
}

type ListPreSpanResp struct {
	Spans loop_span.SpanList
}

type GetTraceReq struct {
	WorkspaceID  int64
	LogID        string
	TraceID      string
	StartTime    int64 // ms
	EndTime      int64 // ms
	PlatformType loop_span.PlatformType
	SpanIDs      []string
	WithDetail   bool
	Filters      *loop_span.FilterFields
}

type GetTraceResp struct {
	TraceId string
	Spans   loop_span.SpanList
}

type SearchTraceOApiReq struct {
	WorkspaceID           int64
	ThirdPartyWorkspaceID string
	Tenants               []string
	TraceID               string
	LogID                 string
	StartTime             int64 // ms
	EndTime               int64 // ms
	Limit                 int32
	SpanIDs               []string
	PlatformType          loop_span.PlatformType
	WithDetail            bool
	Filters               *loop_span.FilterFields
}

type SearchTraceOApiResp struct {
	Spans loop_span.SpanList
}

type ListSpansOApiReq struct {
	WorkspaceID           int64
	ThirdPartyWorkspaceID string
	Tenants               []string
	StartTime             int64 // ms
	EndTime               int64 // ms
	Filters               *loop_span.FilterFields
	Limit                 int32
	DescByStartTime       bool
	PageToken             string
	PlatformType          loop_span.PlatformType
	SpanListType          loop_span.SpanListType
}

type ListSpansOApiResp struct {
	Spans         loop_span.SpanList
	NextPageToken string
	HasMore       bool
}

type TraceQueryParam struct {
	TraceID   string
	StartTime int64 // ms
	EndTime   int64 // ms
}

type GetTracesAdvanceInfoReq struct {
	WorkspaceID           int64
	ThirdPartyWorkspaceID string
	Traces                []*TraceQueryParam
	PlatformType          loop_span.PlatformType
}

type GetTracesAdvanceInfoResp struct {
	Infos []*loop_span.TraceAdvanceInfo
}

type IngestTracesReq struct {
	Tenant           string
	TTL              loop_span.TTL
	WhichIsEnough    int
	CozeAccountId    string
	VolcanoAccountID int64
	Spans            loop_span.SpanList
}

type SendTraceResp struct{}

type GetTracesMetaInfoReq struct {
	WorkspaceID  int64
	PlatformType loop_span.PlatformType
	SpanListType loop_span.SpanListType
}

type GetTracesMetaInfoResp struct {
	FilesMetas      map[string]*config.FieldMeta
	KeySpanTypeList []string
}

type CreateAnnotationReq struct {
	WorkspaceID   int64
	SpanID        string
	TraceID       string
	AnnotationKey string
	AnnotationVal loop_span.AnnotationValue
	Reasoning     string
	QueryDays     int64
	Caller        string
}
type DeleteAnnotationReq struct {
	WorkspaceID   int64
	SpanID        string
	TraceID       string
	AnnotationKey string
	QueryDays     int64
	Caller        string
}

type CreateManualAnnotationReq struct {
	PlatformType loop_span.PlatformType
	Annotation   *loop_span.Annotation
}

type CreateManualAnnotationResp struct {
	AnnotationID string
}

type UpdateManualAnnotationReq struct {
	AnnotationID string
	Annotation   *loop_span.Annotation
	PlatformType loop_span.PlatformType
}

type DeleteManualAnnotationReq struct {
	AnnotationID  string
	WorkspaceID   int64
	TraceID       string
	SpanID        string
	StartTime     int64 // ms
	AnnotationKey string
	PlatformType  loop_span.PlatformType
}

type ListAnnotationsReq struct {
	WorkspaceID     int64
	TraceID         string
	SpanID          string
	StartTime       int64
	DescByUpdatedAt bool
	PlatformType    loop_span.PlatformType
}

type ListAnnotationsResp struct {
	Annotations loop_span.AnnotationList
}

type ChangeEvaluatorScoreRequest struct {
	WorkspaceID  int64
	AnnotationID string
	SpanID       string
	StartTime    int64
	PlatformType loop_span.PlatformType
	Correction   *annotation.Correction
}
type ChangeEvaluatorScoreResp struct {
	Annotation *annotation.Annotation
}
type ListAnnotationEvaluatorsRequest struct {
	WorkspaceID int64
	Name        *string
}
type ListAnnotationEvaluatorsResp struct {
	Evaluators []*annotation.AnnotationEvaluator
}
type ExtractSpanInfoRequest struct {
	WorkspaceID   int64
	TraceID       string
	SpanIds       []string
	StartTime     int64
	EndTime       int64
	PlatformType  loop_span.PlatformType
	FieldMappings []entity.FieldMapping
}
type ExtractSpanInfoResp struct {
	SpanInfos []*trace.SpanInfo
}

type IAnnotationEvent interface {
	Send(ctx context.Context, msg *entity.AnnotationEvent) error
}

//go:generate mockgen -destination=mocks/trace_service.go -package=mocks . ITraceService
type ITraceService interface {
	ListSpans(ctx context.Context, req *ListSpansReq) (*ListSpansResp, error)
	ListPreSpan(ctx context.Context, req *ListPreSpanReq) (r *ListPreSpanResp, err error)
	GetTrace(ctx context.Context, req *GetTraceReq) (*GetTraceResp, error)
	SearchTraceOApi(ctx context.Context, req *SearchTraceOApiReq) (*SearchTraceOApiResp, error)
	ListSpansOApi(ctx context.Context, req *ListSpansOApiReq) (*ListSpansOApiResp, error)
	GetTracesAdvanceInfo(ctx context.Context, req *GetTracesAdvanceInfoReq) (*GetTracesAdvanceInfoResp, error)
	IngestTraces(ctx context.Context, req *IngestTracesReq) error
	GetTracesMetaInfo(ctx context.Context, req *GetTracesMetaInfoReq) (*GetTracesMetaInfoResp, error)
	ListAnnotations(ctx context.Context, req *ListAnnotationsReq) (*ListAnnotationsResp, error)
	CreateAnnotation(ctx context.Context, req *CreateAnnotationReq) error
	DeleteAnnotation(ctx context.Context, req *DeleteAnnotationReq) error
	CreateManualAnnotation(ctx context.Context, req *CreateManualAnnotationReq) (*CreateManualAnnotationResp, error)
	UpdateManualAnnotation(ctx context.Context, req *UpdateManualAnnotationReq) error
	DeleteManualAnnotation(ctx context.Context, req *DeleteManualAnnotationReq) error
	IAnnotationEvent
	ChangeEvaluatorScore(ctx context.Context, req *ChangeEvaluatorScoreRequest) (*ChangeEvaluatorScoreResp, error)
	ListAnnotationEvaluators(ctx context.Context, req *ListAnnotationEvaluatorsRequest) (*ListAnnotationEvaluatorsResp, error)
	ExtractSpanInfo(ctx context.Context, req *ExtractSpanInfoRequest) (*ExtractSpanInfoResp, error)
}

func NewTraceServiceImpl(
	tRepo repo.ITraceRepo,
	traceConfig config.ITraceConfig,
	traceProducer mq.ITraceProducer,
	annotationProducer mq.IAnnotationProducer,
	metrics metrics.ITraceMetrics,
	buildHelper TraceFilterProcessorBuilder,
	tenantProvider tenant.ITenantProvider,
	evalSvc rpc.IEvaluatorRPCAdapter,
	taskRepo taskrepo.ITaskRepo,
	persistentRedis redis.PersistentCmdable,
) (ITraceService, error) {
	return &TraceServiceImpl{
		traceRepo:          tRepo,
		traceConfig:        traceConfig,
		traceProducer:      traceProducer,
		annotationProducer: annotationProducer,
		buildHelper:        buildHelper,
		tenantProvider:     tenantProvider,
		metrics:            metrics,
		evalSvc:            evalSvc,
		taskRepo:           taskRepo,
		persistentRedis:    persistentRedis,
	}, nil
}

type TraceServiceImpl struct {
	traceRepo          repo.ITraceRepo
	traceConfig        config.ITraceConfig
	traceProducer      mq.ITraceProducer
	annotationProducer mq.IAnnotationProducer
	metrics            metrics.ITraceMetrics
	buildHelper        TraceFilterProcessorBuilder
	tenantProvider     tenant.ITenantProvider
	evalSvc            rpc.IEvaluatorRPCAdapter
	taskRepo           taskrepo.ITaskRepo
	persistentRedis    redis.PersistentCmdable
}

const (
	keyPreviousResponseID = "previous_response_id"
	keyResponseID         = "response_id"
)

func (r *TraceServiceImpl) ListPreSpan(ctx context.Context, req *ListPreSpanReq) (resp *ListPreSpanResp, err error) {
	tenants, err := r.getTenants(ctx, req.PlatformType)
	if err != nil {
		return nil, err
	}

	// get pre span ids from redis
	preAndCurrentSpanIDs, respIDByOrder, err := r.traceRepo.GetPreSpanIDs(ctx, &repo.GetPreSpanIDsParam{
		PreRespID: req.PreviousResponseID,
	})
	if err != nil {
		return nil, err
	}
	preAndCurrentSpanIDs = append(preAndCurrentSpanIDs, req.SpanID) // for select current span together

	// batch select from ck
	preAndCurrentSpans, err := r.batchGetPreSpan(ctx, preAndCurrentSpanIDs, tenants, req.StartTime)
	if err != nil {
		return nil, errorx.WrapByCode(err, obErrorx.CommercialCommonInternalErrorCodeCode)
	}

	// span processors
	processors, err := r.buildHelper.BuildListSpansProcessors(ctx, span_processor.Settings{
		WorkspaceId:    req.WorkspaceID,
		PlatformType:   req.PlatformType,
		QueryStartTime: req.StartTime - timeutil.Day2MillSec(30), // past 30 days
		QueryEndTime:   req.StartTime,
		QueryTenants:   tenants,
	})
	if err != nil {
		return nil, errorx.WrapByCode(err, obErrorx.CommercialCommonInternalErrorCodeCode)
	}
	for _, p := range processors {
		preAndCurrentSpans, err = p.Transform(ctx, preAndCurrentSpans)
		if err != nil {
			return nil, err
		}
	}

	// auth check
	if err := r.checkGetPreSpanAuth(ctx, req, tenants, preAndCurrentSpans); err != nil {
		return nil, err
	}

	// order SpanList: remove duplicate span_id, and remove current span
	orderSpans := r.orderPreSpans(preAndCurrentSpans, respIDByOrder)

	return &ListPreSpanResp{Spans: orderSpans}, nil
}

func (r *TraceServiceImpl) batchGetPreSpan(ctx context.Context, spanIDs []string, tenants []string, startTime int64) ([]*loop_span.Span, error) {
	batchNum := 100
	batchPreSpan := make([][]string, 0)
	oneBatchPreSpan := make([]string, 0)
	preAndCurrentSpans := make([]*loop_span.Span, 0)
	for _, spanID := range spanIDs {
		oneBatchPreSpan = append(oneBatchPreSpan, spanID)
		if len(oneBatchPreSpan) == batchNum {
			batchPreSpan = append(batchPreSpan, oneBatchPreSpan)
			oneBatchPreSpan = make([]string, 0)
		}
	}
	if len(oneBatchPreSpan) > 0 {
		batchPreSpan = append(batchPreSpan, oneBatchPreSpan)
	}
	for _, oneBatchSpan := range batchPreSpan {
		dbSpans, err := r.traceRepo.ListSpans(ctx, &repo.ListSpansParam{
			Tenants: tenants,
			Filters: &loop_span.FilterFields{
				FilterFields: []*loop_span.FilterField{
					{
						FieldName: loop_span.SpanFieldSpanId,
						FieldType: loop_span.FieldTypeString,
						Values:    oneBatchSpan,
						QueryType: ptr.Of(loop_span.QueryTypeEnumIn),
					},
				},
			},
			StartAt: startTime - timeutil.Day2MillSec(30), // past 30 days
			EndAt:   startTime + 1,
			Limit:   200,
		})
		if err != nil {
			return nil, err
		}
		if dbSpans != nil && len(dbSpans.Spans) > 0 {
			preAndCurrentSpans = append(preAndCurrentSpans, dbSpans.Spans...)
		}
	}

	return preAndCurrentSpans, nil
}

func (r *TraceServiceImpl) checkGetPreSpanAuth(ctx context.Context, req *ListPreSpanReq, tenants []string, preAndCurrentSpans []*loop_span.Span) error {
	// 1. check current span: check previous_response_id is correct, and if it is in this workspace, pass
	// 2. check pre span: if one span of preSpan in this workspace, pass
	// 3. check span of current trace: if one span of trace in this workspace, pass

	isAuthPass := false
	var currentSpan *loop_span.Span
	for _, span := range preAndCurrentSpans {
		if span.SpanID == req.SpanID && span.TraceID == req.TraceID {
			currentSpan = span
			break
		}
	}
	if currentSpan == nil {
		return errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("current span not found"))
	}
	if preRespID, ok := currentSpan.SystemTagsString[keyPreviousResponseID]; !ok || preRespID != req.PreviousResponseID {
		return errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg(fmt.Sprintf("req previous_response_id is not current span's[%s]", preRespID)))
	}
	if currentSpan.WorkspaceID == strconv.FormatInt(req.WorkspaceID, 10) {
		isAuthPass = true
	}

	if !isAuthPass {
		for _, span := range preAndCurrentSpans {
			if span.WorkspaceID == strconv.FormatInt(req.WorkspaceID, 10) {
				isAuthPass = true
				break
			}
		}
	}

	if !isAuthPass {
		dbSpans, err := r.traceRepo.ListSpans(ctx, &repo.ListSpansParam{
			Tenants: tenants,
			Filters: &loop_span.FilterFields{
				QueryAndOr: ptr.Of(loop_span.QueryAndOrEnumAnd),
				FilterFields: []*loop_span.FilterField{
					{
						FieldName: loop_span.SpanFieldTraceId,
						FieldType: loop_span.FieldTypeString,
						Values:    []string{req.TraceID},
						QueryType: ptr.Of(loop_span.QueryTypeEnumEq),
					},
					{
						FieldName: loop_span.SpanFieldSpaceId,
						FieldType: loop_span.FieldTypeString,
						Values:    []string{strconv.FormatInt(req.WorkspaceID, 10)},
						QueryType: ptr.Of(loop_span.QueryTypeEnumEq),
					},
				},
			},
			StartAt:       req.StartTime - timeutil.Day2MillSec(30), // past 30 days
			EndAt:         req.StartTime,
			SelectColumns: []string{loop_span.SpanFieldSpanId},
			Limit:         1,
		})
		if err != nil {
			return err
		}
		if dbSpans != nil && len(dbSpans.Spans) > 0 {
			isAuthPass = true
		}
	}
	if !isAuthPass {
		return errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("no span in this workspace"))
	}

	return nil
}

func (r *TraceServiceImpl) orderPreSpans(preAndCurrentSpans []*loop_span.Span, respIDByOrder []string) loop_span.SpanList {
	respIDSpanMap := make(map[string]*loop_span.Span)
	for _, span := range preAndCurrentSpans {
		if respID, ok := span.SystemTagsString[keyResponseID]; ok {
			respIDSpanMap[respID] = span
		}
	}
	orderSpans := make(loop_span.SpanList, 0, len(respIDByOrder))
	for i := range respIDByOrder {
		if s, ok := respIDSpanMap[respIDByOrder[i]]; ok {
			orderSpans = append(orderSpans, s)
		}
	}

	return orderSpans
}

func (r *TraceServiceImpl) GetTrace(ctx context.Context, req *GetTraceReq) (*GetTraceResp, error) {
	if req != nil && req.Filters != nil {
		if err := req.Filters.Traverse(processSpecificFilter); err != nil {
			return nil, errorx.WrapByCode(err, obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("invalid filter"))
		}
	}

	tenants, err := r.getTenants(ctx, req.PlatformType)
	if err != nil {
		return nil, err
	}
	omitColumns := make([]string, 0)
	if !req.WithDetail {
		omitColumns = []string{"input", "output"}
	}
	st := time.Now()
	limit := int32(1000)
	if !req.WithDetail {
		limit = 10000
	}
	spans, err := r.traceRepo.GetTrace(ctx, &repo.GetTraceParam{
		Tenants:     tenants,
		LogID:       req.LogID,
		TraceID:     req.TraceID,
		StartAt:     req.StartTime,
		EndAt:       req.EndTime,
		Limit:       limit,
		SpanIDs:     req.SpanIDs,
		Filters:     req.Filters,
		OmitColumns: omitColumns,
	})
	r.metrics.EmitGetTrace(req.WorkspaceID, st, err != nil)
	if err != nil {
		return nil, err
	}
	processors, err := r.buildHelper.BuildGetTraceProcessors(ctx, span_processor.Settings{
		WorkspaceId:     req.WorkspaceID,
		PlatformType:    req.PlatformType,
		QueryStartTime:  req.StartTime,
		QueryEndTime:    req.EndTime,
		SpanDoubleCheck: len(req.SpanIDs) > 0 || (req.Filters != nil && len(req.Filters.FilterFields) > 0),
		QueryTenants:    tenants,
		QueryLogID:      req.LogID,
		QueryTraceID:    req.TraceID,
	})
	if err != nil {
		return nil, errorx.WrapByCode(err, obErrorx.CommercialCommonInternalErrorCodeCode)
	}
	for _, p := range processors {
		spans, err = p.Transform(ctx, spans)
		if err != nil {
			return nil, err
		}
	}
	spans.SortByStartTime(false)
	return &GetTraceResp{
		TraceId: req.TraceID,
		Spans:   spans,
	}, nil
}

func (r *TraceServiceImpl) ListSpans(ctx context.Context, req *ListSpansReq) (*ListSpansResp, error) {
	if err := req.Filters.Traverse(processSpecificFilter); err != nil {
		return nil, errorx.WrapByCode(err, obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("invalid filter"))
	}
	platformFilter, err := r.buildHelper.BuildPlatformRelatedFilter(ctx, req.PlatformType)
	if err != nil {
		return nil, err
	}
	builtinFilter, err := r.buildBuiltinFilters(ctx, platformFilter, req)
	if err != nil {
		return nil, err
	} else if builtinFilter == nil {
		return &ListSpansResp{Spans: loop_span.SpanList{}}, nil
	}
	filters := r.combineFilters(builtinFilter, req.Filters)
	tenants, err := r.getTenants(ctx, req.PlatformType)
	if err != nil {
		return nil, err
	}
	st := time.Now()
	tRes, err := r.traceRepo.ListSpans(ctx, &repo.ListSpansParam{
		Tenants:         tenants,
		Filters:         filters,
		StartAt:         req.StartTime,
		EndAt:           req.EndTime,
		Limit:           req.Limit,
		DescByStartTime: req.DescByStartTime,
		PageToken:       req.PageToken,
	})
	r.metrics.EmitListSpans(req.WorkspaceID, string(req.SpanListType), st, err != nil)
	if err != nil {
		return nil, err
	}
	spans := tRes.Spans
	processors, err := r.buildHelper.BuildListSpansProcessors(ctx, span_processor.Settings{
		WorkspaceId:    req.WorkspaceID,
		PlatformType:   req.PlatformType,
		QueryStartTime: req.StartTime,
		QueryEndTime:   req.EndTime,
		QueryTenants:   tenants,
	})
	if err != nil {
		return nil, errorx.WrapByCode(err, obErrorx.CommercialCommonInternalErrorCodeCode)
	}
	for _, p := range processors {
		spans, err = p.Transform(ctx, spans)
		if err != nil {
			return nil, err
		}
	}
	return &ListSpansResp{
		Spans:         spans,
		NextPageToken: tRes.PageToken,
		HasMore:       tRes.HasMore,
	}, nil
}

func (r *TraceServiceImpl) SearchTraceOApi(ctx context.Context, req *SearchTraceOApiReq) (*SearchTraceOApiResp, error) {
	if req != nil && req.Filters != nil {
		if err := req.Filters.Traverse(processSpecificFilter); err != nil {
			return nil, errorx.WrapByCode(err, obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("invalid filter"))
		}
	}

	omitColumns := make([]string, 0)
	if !req.WithDetail {
		omitColumns = []string{"input", "output"}
	}

	spans, err := r.traceRepo.GetTrace(ctx, &repo.GetTraceParam{
		Tenants:            req.Tenants,
		TraceID:            req.TraceID,
		LogID:              req.LogID,
		SpanIDs:            req.SpanIDs,
		StartAt:            req.StartTime,
		EndAt:              req.EndTime,
		Limit:              req.Limit,
		NotQueryAnnotation: false,
		Filters:            req.Filters,
		OmitColumns:        omitColumns,
	})
	if err != nil {
		return nil, err
	}
	processors, err := r.buildHelper.BuildSearchTraceOApiProcessors(ctx, span_processor.Settings{
		WorkspaceId:           req.WorkspaceID,
		ThirdPartyWorkspaceID: req.ThirdPartyWorkspaceID,
		QueryStartTime:        req.StartTime,
		QueryEndTime:          req.EndTime,
		PlatformType:          req.PlatformType,
		SpanDoubleCheck:       req.Filters != nil && len(req.Filters.FilterFields) > 0,
		QueryTenants:          req.Tenants,
		QueryTraceID:          req.TraceID,
		QueryLogID:            req.LogID,
	})
	if err != nil {
		return nil, errorx.WrapByCode(err, obErrorx.CommercialCommonInternalErrorCodeCode)
	}
	for _, p := range processors {
		spans, err = p.Transform(ctx, spans)
		if err != nil {
			return nil, err
		}
	}
	spans.SortByStartTime(false)
	return &SearchTraceOApiResp{
		Spans: spans,
	}, nil
}

func (r *TraceServiceImpl) ListSpansOApi(ctx context.Context, req *ListSpansOApiReq) (*ListSpansOApiResp, error) {
	if err := req.Filters.Traverse(processSpecificFilter); err != nil {
		return nil, errorx.WrapByCode(err, obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("invalid filter"))
	}
	platformFilter, err := r.buildHelper.BuildPlatformRelatedFilter(ctx, req.PlatformType)
	if err != nil {
		return nil, err
	}
	builtinFilter, err := r.buildBuiltinFilters(ctx, platformFilter, &ListSpansReq{
		WorkspaceID:           req.WorkspaceID,
		ThirdPartyWorkspaceID: req.ThirdPartyWorkspaceID,
		SpanListType:          req.SpanListType,
	})
	if err != nil {
		return nil, err
	} else if builtinFilter == nil {
		return &ListSpansOApiResp{Spans: loop_span.SpanList{}}, nil
	}
	filters := r.combineFilters(builtinFilter, req.Filters)
	tRes, err := r.traceRepo.ListSpans(ctx, &repo.ListSpansParam{
		Tenants:         req.Tenants,
		Filters:         filters,
		StartAt:         req.StartTime,
		EndAt:           req.EndTime,
		Limit:           req.Limit,
		DescByStartTime: req.DescByStartTime,
		PageToken:       req.PageToken,
	})
	if err != nil {
		return nil, err
	}

	spans := tRes.Spans
	processors, err := r.buildHelper.BuildListSpansOApiProcessors(ctx, span_processor.Settings{
		WorkspaceId:    req.WorkspaceID,
		QueryStartTime: req.StartTime,
		QueryEndTime:   req.EndTime,
		QueryTenants:   req.Tenants,
	})
	if err != nil {
		return nil, errorx.WrapByCode(err, obErrorx.CommercialCommonInternalErrorCodeCode)
	}
	for _, p := range processors {
		spans, err = p.Transform(ctx, spans)
		if err != nil {
			return nil, err
		}
	}
	return &ListSpansOApiResp{
		Spans:         spans,
		NextPageToken: tRes.PageToken,
		HasMore:       tRes.HasMore,
	}, nil
}

func (r *TraceServiceImpl) IngestTraces(ctx context.Context, req *IngestTracesReq) error {
	processors, err := r.buildHelper.BuildIngestTraceProcessors(ctx, span_processor.Settings{})
	if err != nil {
		return errorx.WrapByCode(err, obErrorx.CommercialCommonInternalErrorCodeCode)
	}
	for _, p := range processors {
		req.Spans, err = p.Transform(ctx, req.Spans)
		if err != nil {
			return err
		}
	}

	traceData := &entity.TraceData{
		Tenant: req.Tenant,
		TenantInfo: entity.TenantInfo{
			TTL:              req.TTL,
			WorkspaceId:      req.Spans[0].WorkspaceID,
			CozeAccountID:    req.CozeAccountId,
			WhichIsEnough:    req.WhichIsEnough,
			VolcanoAccountID: req.VolcanoAccountID,
		},
		SpanList: req.Spans,
	}
	if err := r.traceProducer.IngestSpans(ctx, traceData); err != nil {
		return err
	}
	logs.CtxInfo(ctx, "Send msg successfully, spans count %d", len(req.Spans))
	return nil
}

func (r *TraceServiceImpl) GetTracesAdvanceInfo(ctx context.Context, req *GetTracesAdvanceInfoReq) (*GetTracesAdvanceInfoResp, error) {
	var (
		g                errgroup.Group
		lock             sync.Mutex
		defaultTimeRange = int64(60 * 60 * 1000) // ms
	)
	tenants, err := r.getTenants(ctx, req.PlatformType)
	if err != nil {
		return nil, err
	}
	resp := &GetTracesAdvanceInfoResp{
		Infos: []*loop_span.TraceAdvanceInfo{},
	}
	for _, v := range req.Traces {
		g.Go(func() error {
			defer goroutine.Recovery(ctx)
			qReq := &repo.GetTraceParam{
				Tenants:            tenants,
				TraceID:            v.TraceID,
				StartAt:            v.StartTime,
				EndAt:              v.EndTime + defaultTimeRange,
				Limit:              1000,
				NotQueryAnnotation: true, // no need to query annotation
				OmitColumns: []string{
					loop_span.SpanFieldInput,
					loop_span.SpanFieldOutput,
				},
				Filters: loop_span.GetModelSpansFilter(),
			}
			st := time.Now()
			spans, err := r.traceRepo.GetTrace(ctx, qReq)
			r.metrics.EmitGetTrace(req.WorkspaceID, st, err != nil)
			if err != nil {
				logs.CtxError(ctx, "Fail to get trace %v, %v", *qReq, err)
				return err
			}
			processors, err := r.buildHelper.BuildAdvanceInfoProcessors(ctx, span_processor.Settings{
				WorkspaceId:     req.WorkspaceID,
				PlatformType:    req.PlatformType,
				QueryStartTime:  v.StartTime,
				QueryEndTime:    v.EndTime + defaultTimeRange,
				SpanDoubleCheck: true,
				QueryTenants:    tenants,
				QueryTraceID:    v.TraceID,
			})
			if err != nil {
				logs.CtxError(ctx, "Fail to build advance info processor, %v", err)
				return err
			}
			for _, p := range processors {
				spans, err = p.Transform(ctx, spans)
				if err != nil {
					logs.CtxWarn(ctx, "Fail to transform span, %v", err)
					return nil
				}
			}
			inputTokens, outputTokens, err := spans.Stat(ctx)
			if err != nil {
				logs.CtxWarn(ctx, "Fail to get spans stat, %v", err)
				return nil
			}
			lock.Lock()
			defer lock.Unlock()
			resp.Infos = append(resp.Infos, &loop_span.TraceAdvanceInfo{
				TraceId:    qReq.TraceID,
				InputCost:  inputTokens,
				OutputCost: outputTokens,
			})
			return nil
		})
	}
	if err := g.Wait(); err != nil {
		logs.CtxError(ctx, "fail to get all trace advance info, %v", err)
		return nil, err
	}
	return resp, nil
}

func (r *TraceServiceImpl) GetTracesMetaInfo(ctx context.Context, req *GetTracesMetaInfoReq) (*GetTracesMetaInfoResp, error) {
	cfg, err := r.traceConfig.GetTraceFieldMetaInfo(ctx)
	if err != nil {
		return nil, errorx.WrapByCode(err, obErrorx.CommercialCommonInternalErrorCodeCode)
	}
	baseFields, ok := cfg.FieldMetas[loop_span.PlatformDefault][req.SpanListType]
	if !ok {
		return nil, errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("base meta info not found"))
	}

	fields, ok := cfg.FieldMetas[req.PlatformType][req.SpanListType]
	if !ok {
		logs.CtxWarn(ctx, "FieldMetas not found: %v-%v", req.PlatformType, req.SpanListType)
	}
	fieldMetas := make(map[string]*config.FieldMeta)
	for _, field := range baseFields {
		fieldMta, ok := cfg.AvailableFields[field]
		if !ok || fieldMta == nil {
			logs.CtxError(ctx, "GetTracesMetaInfo invalid field: %v", field)
			return nil, errorx.NewByCode(obErrorx.CommercialCommonInternalErrorCodeCode)
		}
		fieldMetas[field] = fieldMta
	}
	for _, field := range fields {
		fieldMta, ok := cfg.AvailableFields[field]
		if !ok || fieldMta == nil {
			logs.CtxError(ctx, "GetTracesMetaInfo invalid field: %v", field)
			return nil, errorx.NewByCode(obErrorx.CommercialCommonInternalErrorCodeCode)
		}
		fieldMetas[field] = fieldMta
	}

	spanTypeCfg := r.traceConfig.GetKeySpanTypes(ctx)
	keySpanTypes, ok := spanTypeCfg[string(req.PlatformType)]
	if !ok {
		keySpanTypes = spanTypeCfg[string(loop_span.PlatformDefault)]
	}
	return &GetTracesMetaInfoResp{
		FilesMetas:      fieldMetas,
		KeySpanTypeList: keySpanTypes,
	}, nil
}

func (r *TraceServiceImpl) ListAnnotations(ctx context.Context, req *ListAnnotationsReq) (*ListAnnotationsResp, error) {
	tenants, err := r.getTenants(ctx, req.PlatformType)
	if err != nil {
		return nil, err
	}
	annotations, err := r.traceRepo.ListAnnotations(ctx, &repo.ListAnnotationsParam{
		Tenants:         tenants,
		SpanID:          req.SpanID,
		TraceID:         req.TraceID,
		WorkspaceId:     req.WorkspaceID,
		DescByUpdatedAt: req.DescByUpdatedAt,
		StartAt:         req.StartTime - time.Second.Milliseconds(),
		EndAt:           req.StartTime + time.Second.Milliseconds(),
	})
	if err != nil {
		return nil, err
	}
	return &ListAnnotationsResp{
		Annotations: annotations,
	}, nil
}

func (r *TraceServiceImpl) CreateManualAnnotation(ctx context.Context, req *CreateManualAnnotationReq) (*CreateManualAnnotationResp, error) {
	tenants, err := r.getTenants(ctx, req.PlatformType)
	if err != nil {
		return nil, err
	}
	spans, err := r.getSpan(ctx,
		tenants,
		[]string{req.Annotation.SpanID},
		req.Annotation.TraceID,
		req.Annotation.WorkspaceID,
		req.Annotation.StartTime.Add(-time.Second).UnixMilli(),
		req.Annotation.StartTime.Add(time.Second).UnixMilli(),
	)
	if err != nil {
		return nil, err
	} else if len(spans) == 0 {
		logs.CtxWarn(ctx, "no span found for span_id %s trace_id %s", req.Annotation.SpanID, req.Annotation.TraceID)
		return nil, errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode)
	}
	span := spans[0]
	annotation, err := span.BuildFeedback(
		loop_span.AnnotationTypeManualFeedback,
		req.Annotation.Key,
		req.Annotation.Value,
		req.Annotation.Reasoning,
		session.UserIDInCtxOrEmpty(ctx),
		false,
	)
	if err != nil {
		return nil, errorx.WrapByCode(err, obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("invalid annotation"))
	}
	if err := r.traceRepo.InsertAnnotations(ctx, &repo.InsertAnnotationParam{
		Tenant:         span.GetTenant(),
		TTL:            span.GetTTL(ctx),
		Span:           span,
		AnnotationType: gptr.Of(annotation.AnnotationType),
	}); err != nil {
		return nil, err
	}
	return &CreateManualAnnotationResp{
		AnnotationID: annotation.ID,
	}, nil
}

func (r *TraceServiceImpl) UpdateManualAnnotation(ctx context.Context, req *UpdateManualAnnotationReq) error {
	tenants, err := r.getTenants(ctx, req.PlatformType)
	if err != nil {
		return err
	}
	spans, err := r.getSpan(ctx,
		tenants,
		[]string{req.Annotation.SpanID},
		req.Annotation.TraceID,
		req.Annotation.WorkspaceID,
		req.Annotation.StartTime.Add(-time.Second).UnixMilli(),
		req.Annotation.StartTime.Add(time.Second).UnixMilli(),
	)
	if err != nil {
		return err
	} else if len(spans) == 0 {
		logs.CtxWarn(ctx, "no span found for span_id %s trace_id %s", req.Annotation.SpanID, req.Annotation.TraceID)
		return errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode)
	}
	span := spans[0]
	annotation, err := span.BuildFeedback(
		loop_span.AnnotationTypeManualFeedback,
		req.Annotation.Key,
		req.Annotation.Value,
		req.Annotation.Reasoning,
		session.UserIDInCtxOrEmpty(ctx),
		false,
	)
	if err != nil || annotation.ID != req.AnnotationID {
		return errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode)
	}
	existedAnno, err := r.traceRepo.GetAnnotation(ctx, &repo.GetAnnotationParam{
		Tenants: tenants,
		ID:      req.AnnotationID,
		StartAt: time.UnixMicro(span.StartTime).Add(-time.Second).UnixMilli(),
		EndAt:   time.UnixMicro(span.StartTime).Add(time.Second).UnixMilli(),
	})
	if err != nil {
		logs.CtxError(ctx, "get annotation %s err %v", req.AnnotationID, err)
		return err
	} else if existedAnno != nil {
		annotation.CreatedBy = existedAnno.CreatedBy
		annotation.CreatedAt = existedAnno.CreatedAt
	}
	return r.traceRepo.InsertAnnotations(ctx, &repo.InsertAnnotationParam{
		Tenant:         span.GetTenant(),
		TTL:            span.GetTTL(ctx),
		Span:           span,
		AnnotationType: gptr.Of(annotation.AnnotationType),
	})
}

func (r *TraceServiceImpl) DeleteManualAnnotation(ctx context.Context, req *DeleteManualAnnotationReq) error {
	tenants, err := r.getTenants(ctx, req.PlatformType)
	if err != nil {
		return err
	}
	spans, err := r.getSpan(ctx,
		tenants,
		[]string{req.SpanID},
		req.TraceID,
		strconv.FormatInt(req.WorkspaceID, 10),
		req.StartTime-time.Second.Milliseconds(),
		req.StartTime+time.Second.Milliseconds(),
	)
	if err != nil {
		return err
	} else if len(spans) == 0 {
		logs.CtxWarn(ctx, "no span found for span_id %s trace_id %s", req.SpanID, req.TraceID)
		return errorx.NewByCode(obErrorx.CommercialCommonInternalErrorCodeCode)
	}
	span := spans[0]
	annotation, err := span.BuildFeedback(
		loop_span.AnnotationTypeManualFeedback,
		req.AnnotationKey,
		loop_span.AnnotationValue{},
		"",
		session.UserIDInCtxOrEmpty(ctx),
		true,
	)
	if err != nil || annotation.ID != req.AnnotationID {
		return errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("invalid annotation"))
	}
	return r.traceRepo.InsertAnnotations(ctx, &repo.InsertAnnotationParam{
		Tenant:         span.GetTenant(),
		TTL:            span.GetTTL(ctx),
		Span:           span,
		AnnotationType: gptr.Of(annotation.AnnotationType),
	})
}

func (r *TraceServiceImpl) CreateAnnotation(ctx context.Context, req *CreateAnnotationReq) error {
	cfg, err := r.getAnnotationCallerCfg(ctx, req.Caller)
	if err != nil {
		return err
	}
	spans, err := r.getSpan(ctx,
		cfg.Tenants,
		[]string{req.SpanID},
		req.TraceID,
		strconv.FormatInt(req.WorkspaceID, 10),
		time.Now().Add(-time.Duration(req.QueryDays)*24*time.Hour).UnixMilli(),
		time.Now().UnixMilli(),
	)
	if err != nil {
		return err
	} else if len(spans) == 0 {
		return r.annotationProducer.SendAnnotation(ctx, &entity.AnnotationEvent{
			Annotation: &loop_span.Annotation{
				SpanID:         req.SpanID,
				TraceID:        req.TraceID,
				WorkspaceID:    strconv.FormatInt(req.WorkspaceID, 10),
				AnnotationType: loop_span.AnnotationType(cfg.AnnotationType),
				Key:            req.AnnotationKey,
				Value:          req.AnnotationVal,
				Reasoning:      req.Reasoning,
				Status:         loop_span.AnnotationStatusNormal,
				CreatedAt:      time.Now(),
				UpdatedAt:      time.Now(),
			},
			Caller:     req.Caller,
			StartAt:    time.Now().Add(-24 * time.Hour).UnixMilli(),
			EndAt:      time.Now().Add(1 * time.Hour).UnixMilli(),
			RetryTimes: 3,
		})
	}
	span := spans[0]
	annotation, err := span.BuildFeedback(
		loop_span.AnnotationType(cfg.AnnotationType),
		req.AnnotationKey,
		req.AnnotationVal,
		req.Reasoning, "", false,
	)
	if err != nil {
		return errorx.WrapByCode(err, obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("invalid annotation"))
	}
	existedAnno, err := r.traceRepo.GetAnnotation(ctx, &repo.GetAnnotationParam{
		Tenants: cfg.Tenants,
		ID:      annotation.ID,
		StartAt: time.UnixMicro(span.StartTime).Add(-time.Second).UnixMilli(),
		EndAt:   time.UnixMicro(span.StartTime).Add(time.Second).UnixMilli(),
	})
	if err != nil {
		return err
	} else if existedAnno != nil {
		annotation.CreatedBy = existedAnno.CreatedBy
		annotation.CreatedAt = existedAnno.CreatedAt
	}
	return r.traceRepo.InsertAnnotations(ctx, &repo.InsertAnnotationParam{
		Tenant:         span.GetTenant(),
		TTL:            span.GetTTL(ctx),
		Span:           span,
		AnnotationType: gptr.Of(annotation.AnnotationType),
	})
}

func (r *TraceServiceImpl) DeleteAnnotation(ctx context.Context, req *DeleteAnnotationReq) error {
	cfg, err := r.getAnnotationCallerCfg(ctx, req.Caller)
	if err != nil {
		return err
	}
	spans, err := r.getSpan(ctx,
		cfg.Tenants,
		[]string{req.SpanID},
		req.TraceID,
		strconv.FormatInt(req.WorkspaceID, 10),
		time.Now().Add(-time.Duration(req.QueryDays)*24*time.Hour).UnixMilli(),
		time.Now().UnixMilli(),
	)
	if err != nil {
		return err
	} else if len(spans) == 0 {
		return r.annotationProducer.SendAnnotation(ctx, &entity.AnnotationEvent{
			Annotation: &loop_span.Annotation{
				SpanID:         req.SpanID,
				TraceID:        req.TraceID,
				WorkspaceID:    strconv.FormatInt(req.WorkspaceID, 10),
				AnnotationType: loop_span.AnnotationType(cfg.AnnotationType),
				Key:            req.AnnotationKey,
				Status:         loop_span.AnnotationStatusDeleted,
				CreatedAt:      time.Now(),
				UpdatedAt:      time.Now(),
				IsDeleted:      true,
			},
			Caller:     req.Caller,
			StartAt:    time.Now().Add(-24 * time.Hour).UnixMilli(),
			EndAt:      time.Now().Add(1 * time.Hour).UnixMilli(),
			RetryTimes: 3,
		})
	}
	span := spans[0]
	annotation, err := span.BuildFeedback(
		loop_span.AnnotationType(cfg.AnnotationType),
		req.AnnotationKey,
		loop_span.AnnotationValue{}, "", "",
		true,
	)
	if err != nil {
		return errorx.WrapByCode(err, obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("invalid annotation"))
	}
	return r.traceRepo.InsertAnnotations(ctx, &repo.InsertAnnotationParam{
		Tenant:         span.GetTenant(),
		TTL:            span.GetTTL(ctx),
		Span:           span,
		AnnotationType: gptr.Of(annotation.AnnotationType),
	})
}

func (r *TraceServiceImpl) Send(ctx context.Context, event *entity.AnnotationEvent) error {
	shouldReSend := false
	defer func() {
		event.RetryTimes--
		// resend if not success
		if !shouldReSend || event.RetryTimes <= 0 {
			return
		}
		logs.CtxInfo(ctx, "resend annotation event")
		_ = r.annotationProducer.SendAnnotation(ctx, event)
	}()
	cfg, err := r.getAnnotationCallerCfg(ctx, event.Caller)
	if err != nil { // retry
		return err
	}
	spans, err := r.getSpan(ctx,
		cfg.Tenants,
		[]string{event.Annotation.SpanID},
		event.Annotation.TraceID,
		event.Annotation.WorkspaceID,
		event.StartAt,
		event.EndAt,
	)
	if err != nil || len(spans) == 0 { // retry if not found yet
		shouldReSend = true
		return nil
	}
	span := spans[0]
	event.Annotation.StartTime = time.UnixMicro(span.StartTime)
	if err := event.Annotation.GenID(); err != nil {
		logs.CtxWarn(ctx, "failed to generate annotation id for %+v, %v", event.Annotation, err)
		return nil
	}
	// retry if failed
	return r.traceRepo.InsertAnnotations(ctx, &repo.InsertAnnotationParam{
		Tenant:         span.GetTenant(),
		TTL:            span.GetTTL(ctx),
		Span:           span,
		AnnotationType: gptr.Of(event.Annotation.AnnotationType),
	})
}

func (r *TraceServiceImpl) getSpan(ctx context.Context, tenants []string, spanIds []string, traceId, workspaceId string, startAt, endAt int64) ([]*loop_span.Span, error) {
	validSpanIds := make([]string, 0, len(spanIds))
	for _, span := range spanIds {
		if span == "" {
			continue
		}
		validSpanIds = append(validSpanIds, span)
	}
	if (len(validSpanIds) == 0 && traceId == "") || workspaceId == "" {
		return nil, errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode)
	}
	var filterFields []*loop_span.FilterField
	if len(validSpanIds) != 0 {
		filterFields = append(filterFields,
			&loop_span.FilterField{
				FieldName: loop_span.SpanFieldSpanId,
				FieldType: loop_span.FieldTypeString,
				Values:    validSpanIds,
				QueryType: ptr.Of(loop_span.QueryTypeEnumIn),
			})
	} else {
		filterFields = append(filterFields,
			&loop_span.FilterField{
				FieldName: loop_span.SpanFieldParentID,
				FieldType: loop_span.FieldTypeString,
				Values:    []string{"0", ""},
				QueryType: ptr.Of(loop_span.QueryTypeEnumIn),
			})
	}
	filterFields = append(filterFields, &loop_span.FilterField{
		FieldName: loop_span.SpanFieldSpaceId,
		FieldType: loop_span.FieldTypeString,
		Values:    []string{workspaceId},
		QueryType: ptr.Of(loop_span.QueryTypeEnumEq),
	})

	if traceId != "" {
		filterFields = append(filterFields, &loop_span.FilterField{
			FieldName: loop_span.SpanFieldTraceId,
			FieldType: loop_span.FieldTypeString,
			Values:    []string{traceId},
			QueryType: ptr.Of(loop_span.QueryTypeEnumEq),
		})
	}
	res, err := r.traceRepo.ListSpans(ctx, &repo.ListSpansParam{
		Tenants: tenants,
		Filters: &loop_span.FilterFields{
			FilterFields: filterFields,
		},
		StartAt:            startAt,
		EndAt:              endAt,
		NotQueryAnnotation: true,
		Limit:              2,
	})
	if err != nil {
		logs.CtxError(ctx, "failed to list span, %v", err)
		return nil, err
	} else if len(res.Spans) == 0 {
		return nil, nil
	}
	return res.Spans, nil
}

func (r *TraceServiceImpl) getAnnotationCallerCfg(ctx context.Context, caller string) (*config.AnnotationConfig, error) {
	cfg, err := r.traceConfig.GetAnnotationSourceCfg(ctx)
	if err != nil {
		return nil, err
	}
	callerCfg, ok := cfg.SourceCfg[caller]
	if ok {
		return &callerCfg, nil
	}
	callerCfg, ok = cfg.SourceCfg["default"]
	if ok {
		return &callerCfg, nil
	}
	return nil, errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode)
}

func (r *TraceServiceImpl) buildBuiltinFilters(ctx context.Context, f span_filter.Filter, req *ListSpansReq) (*loop_span.FilterFields, error) {
	filters := make([]*loop_span.FilterField, 0)
	env := &span_filter.SpanEnv{
		WorkspaceID:           req.WorkspaceID,
		ThirdPartyWorkspaceID: req.ThirdPartyWorkspaceID,
	}
	basicFilter, forceQuery, err := f.BuildBasicSpanFilter(ctx, env)
	if err != nil {
		return nil, err
	} else if len(basicFilter) == 0 && !forceQuery { // if it's null, no need to query from ck
		return nil, nil
	}
	filters = append(filters, basicFilter...)
	switch req.SpanListType {
	case loop_span.SpanListTypeRootSpan:
		subFilter, err := f.BuildRootSpanFilter(ctx, env)
		if err != nil {
			return nil, err
		}
		filters = append(filters, subFilter...)
	case loop_span.SpanListTypeLLMSpan:
		subFilter, err := f.BuildLLMSpanFilter(ctx, env)
		if err != nil {
			return nil, err
		}
		filters = append(filters, subFilter...)
	case loop_span.SpanListTypeAllSpan:
		subFilter, err := f.BuildALLSpanFilter(ctx, env)
		if err != nil {
			return nil, err
		}
		filters = append(filters, subFilter...)
	default:
		return nil, errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("invalid span list type: %s"))
	}
	filterAggr := &loop_span.FilterFields{
		QueryAndOr:   ptr.Of(loop_span.QueryAndOrEnumAnd),
		FilterFields: filters,
	}
	return filterAggr, nil
}

func (r *TraceServiceImpl) combineFilters(filters ...*loop_span.FilterFields) *loop_span.FilterFields {
	filterAggr := &loop_span.FilterFields{
		QueryAndOr: ptr.Of(loop_span.QueryAndOrEnumAnd),
	}
	for _, f := range filters {
		if f == nil {
			continue
		}
		filterAggr.FilterFields = append(filterAggr.FilterFields, &loop_span.FilterField{
			QueryAndOr: ptr.Of(loop_span.QueryAndOrEnumAnd),
			SubFilter:  f,
		})
	}
	return filterAggr
}

func (r *TraceServiceImpl) getTenants(ctx context.Context, platform loop_span.PlatformType) ([]string, error) {
	return r.tenantProvider.GetTenantsByPlatformType(ctx, platform)
}

func (r *TraceServiceImpl) ChangeEvaluatorScore(ctx context.Context, req *ChangeEvaluatorScoreRequest) (*ChangeEvaluatorScoreResp, error) {
	var resp *ChangeEvaluatorScoreResp
	tenants, err := r.getTenants(ctx, req.PlatformType)
	if err != nil {
		return resp, err
	}
	spans, err := r.getSpan(ctx,
		tenants,
		[]string{req.SpanID},
		"",
		strconv.FormatInt(req.WorkspaceID, 10),
		req.StartTime-time.Second.Milliseconds(),
		req.StartTime+time.Second.Milliseconds(),
	)
	if err != nil {
		return resp, err
	} else if len(spans) == 0 {
		logs.CtxWarn(ctx, "no span found for span_id %s", req.SpanID)
		return resp, errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode)
	}
	span := spans[0]
	annotation, err := r.traceRepo.GetAnnotation(ctx, &repo.GetAnnotationParam{
		Tenants: tenants,
		ID:      req.AnnotationID,
		StartAt: time.UnixMicro(span.StartTime).Add(-time.Second).UnixMilli(),
		EndAt:   time.UnixMicro(span.StartTime).Add(time.Second).UnixMilli(),
	})
	if err != nil {
		logs.CtxError(ctx, "get annotation %s err %v", req.AnnotationID, err)
		return resp, errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("get annotation error"))
	}
	if annotation == nil {
		return resp, errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("annotation not found"))
	}
	updateBy := session.UserIDInCtxOrEmpty(ctx)
	if updateBy == "" {
		return resp, errorx.NewByCode(obErrorx.UserParseFailedCode)
	}
	annotation.CorrectAutoEvaluateScore(req.Correction.GetScore(), req.Correction.GetExplain(), updateBy)
	// 以评估数据为主数据，优先修改评估数据，异常则直接返回失败
	if err = r.correctEvaluatorRecords(ctx, r.evalSvc, annotation); err != nil {
		return resp, err
	}
	// 再同步修改观测数据
	span.Annotations = append(span.Annotations, annotation)
	param := &repo.InsertAnnotationParam{
		Tenant:         span.GetTenant(),
		TTL:            span.GetTTL(ctx),
		Span:           span,
		AnnotationType: gptr.Of(annotation.AnnotationType),
	}
	if err = r.traceRepo.InsertAnnotations(ctx, param); err != nil {
		recordID := lo.Ternary(annotation.GetAutoEvaluateMetadata() != nil, annotation.GetAutoEvaluateMetadata().EvaluatorRecordID, 0)
		// 如果同步修改失败，异步补偿
		// todo 异步有问题，会重复
		logs.CtxWarn(ctx, "Sync upsert annotation failed, try async upsert. span_id=[%v], recored_id=[%v], err:%v",
			annotation.SpanID, recordID, err)
		return resp, nil
	}
	return &ChangeEvaluatorScoreResp{
		Annotation: annotation.ToFornaxAnnotation(ctx),
	}, nil
}

func (r *TraceServiceImpl) correctEvaluatorRecords(ctx context.Context, evalSvc rpc.IEvaluatorRPCAdapter, annotation *loop_span.Annotation) error {
	if annotation == nil {
		return errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("annotation is nil"))
	}
	if annotation.GetAutoEvaluateMetadata() == nil {
		return errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("annotation auto evaluate metadata is nil"))
	}
	if len(annotation.Corrections) == 0 {
		return errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("annotation corrections is empty"))
	}
	correction := annotation.Corrections[len(annotation.Corrections)-1]

	if err := evalSvc.UpdateEvaluatorRecord(ctx, &rpc.UpdateEvaluatorRecordParam{
		WorkspaceID:       annotation.WorkspaceID,
		EvaluatorRecordID: annotation.GetAutoEvaluateMetadata().EvaluatorRecordID,
		Score:             correction.Value.FloatValue,
		Reasoning:         correction.Reasoning,
		UpdatedBy:         correction.UpdatedBy,
	}); err != nil {
		return err
	}
	return nil
}

func (r *TraceServiceImpl) ListAnnotationEvaluators(ctx context.Context, req *ListAnnotationEvaluatorsRequest) (*ListAnnotationEvaluatorsResp, error) {
	resp := &ListAnnotationEvaluatorsResp{}
	resp.Evaluators = make([]*annotation.AnnotationEvaluator, 0)
	evaluators := make([]*rpc.Evaluator, 0)

	if req.Name != nil {
		// 有name直接模糊查询
		evaluatorList, err := r.evalSvc.ListEvaluators(ctx, &rpc.ListEvaluatorsParam{
			WorkspaceID: req.WorkspaceID,
			Name:        req.Name,
		})
		if err != nil {
			return resp, err
		}
		evaluators = append(evaluators, evaluatorList...)
	} else {
		// 没有name先查task
		taskDOs, _, err := r.taskRepo.ListTasks(ctx, taskrepo.ListTaskParam{
			WorkspaceIDs: []int64{req.WorkspaceID},
			ReqLimit:     int32(500),
			ReqOffset:    int32(0),
		})
		if err != nil {
			return nil, err
		}
		if len(taskDOs) == 0 {
			logs.CtxInfo(ctx, "GetTasks tasks is nil")
			return resp, nil
		}

		evaluatorVersionIDS := make(map[int64]bool)
		for _, taskDO := range taskDOs {
			taskConfig := tconv.TaskConfigDO2DTO(taskDO.TaskConfig)
			if taskConfig == nil {
				continue
			}
			for _, evaluator := range taskConfig.AutoEvaluateConfigs {
				evaluatorVersionIDS[evaluator.EvaluatorVersionID] = true
				if len(evaluatorVersionIDS) >= 30 {
					break
				}
			}
			if len(evaluatorVersionIDS) >= 30 {
				break
			}
		}
		evaluatorVersionIDList := make([]int64, 0)
		for k := range evaluatorVersionIDS {
			evaluatorVersionIDList = append(evaluatorVersionIDList, k)
		}
		evaluatorList, _, err := r.evalSvc.BatchGetEvaluatorVersions(ctx, &rpc.BatchGetEvaluatorVersionsParam{
			WorkspaceID:         req.WorkspaceID,
			EvaluatorVersionIds: evaluatorVersionIDList,
		})
		if err != nil {
			return resp, errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithMsgParam("evaluatorVersionIDs is invalid, BatchGetEvaluators err: %v", err.Error()))
		}
		evaluators = append(evaluators, evaluatorList...)
	}
	for _, evaluator := range evaluators {
		re := &annotation.AnnotationEvaluator{}
		if evaluator.EvaluatorVersionID != 0 {
			re.EvaluatorVersionID = evaluator.EvaluatorVersionID
		}
		if evaluator.EvaluatorName != "" {
			re.EvaluatorName = evaluator.EvaluatorName
		}
		if evaluator.EvaluatorVersion != "" {
			re.EvaluatorVersion = evaluator.EvaluatorVersion
		}
		resp.Evaluators = append(resp.Evaluators, re)
	}
	return resp, nil
}

func (r *TraceServiceImpl) ExtractSpanInfo(ctx context.Context, req *ExtractSpanInfoRequest) (*ExtractSpanInfoResp, error) {
	resp := &ExtractSpanInfoResp{}
	var spanInfos []*trace.SpanInfo
	tenants, err := r.getTenants(ctx, req.PlatformType)
	if err != nil {
		return resp, err
	}
	spans, err := r.getSpan(ctx,
		tenants,
		req.SpanIds,
		req.TraceID,
		strconv.FormatInt(req.WorkspaceID, 10),
		req.StartTime-time.Second.Milliseconds(),
		req.EndTime+time.Second.Milliseconds(),
	)
	if err != nil {
		return resp, err
	} else if len(spans) == 0 {
		logs.CtxWarn(ctx, "no span found for span_ids %v trace_id %s", req.SpanIds, req.TraceID)
		return resp, errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode)
	}
	logs.CtxInfo(ctx, "Get spans success, total conut:%v", len(spans))
	for _, span := range spans {
		var fieldList []*dataset.FieldData
		for _, mapping := range req.FieldMappings {
			value, err := buildExtractSpanInfo(ctx, span, &mapping)
			if err != nil {
				// 非json但使用了jsonpath，也不报错，置空
				logs.CtxInfo(ctx, "Extract field failed, err:%v", err)
				return resp, err
			}
			content := buildContent(value)
			// 前端传入的是Name，评测集需要的是key，需要做一下mapping
			if mapping.FieldSchema.Name == "" {
				logs.CtxInfo(ctx, "Evaluator field name is nil")
				continue
			}
			fieldList = append(fieldList, &dataset.FieldData{
				Key:     mapping.FieldSchema.Key,
				Name:    gptr.Of(mapping.FieldSchema.Name),
				Content: content,
			})
		}
		spanInfos = append(spanInfos, &trace.SpanInfo{
			SpanID:    span.SpanID,
			FieldList: fieldList,
		})
	}
	return &ExtractSpanInfoResp{
		SpanInfos: spanInfos,
	}, nil
}

func buildExtractSpanInfo(ctx context.Context, span *loop_span.Span, fieldMapping *entity.FieldMapping) (string, error) {
	value, err := span.ExtractByJsonpath(ctx, fieldMapping.TraceFieldKey, fieldMapping.TraceFieldJsonpath)
	if err != nil {
		// 非json但使用了jsonpath，也不报错，置空
		logs.CtxInfo(ctx, "Extract field failed, err:%v", err)
	}
	content, errCode := entity.GetContentInfo(ctx, fieldMapping.FieldSchema.ContentType, value)
	if errCode == entity.DatasetErrorType_MismatchSchema {
		logs.CtxInfo(ctx, "invalid multi part")
		return "", errorx.NewByCode(obErrorx.CommercialCommonInvalidParamCodeCode, errorx.WithExtraMsg("invalid multi part"))
	}
	valueJSON, err := json.Marshal(content)
	if err != nil {
		return "", err
	}
	return string(valueJSON), nil
}

func buildContent(value string) *dataset.Content {
	var content *dataset.Content
	err := json.Unmarshal([]byte(value), &content)
	if err != nil {
		content = &dataset.Content{
			ContentType: gptr.Of(common.ContentTypeText),
			Text:        gptr.Of(value),
		}
	}
	return content
}

func processSpecificFilter(f *loop_span.FilterField) error {
	switch f.FieldName {
	case loop_span.SpanFieldStatus:
		if err := processStatusFilter(f); err != nil {
			return err
		}
	case loop_span.SpanFieldDuration,
		loop_span.SpanFieldLatencyFirstResp,
		loop_span.SpanFieldStartTimeFirstResp,
		loop_span.SpanFieldStartTimeFirstTokenResp,
		loop_span.SpanFieldLatencyFirstTokenResp,
		loop_span.SpanFieldReasoningDuration:
		if err := processLatencyFilter(f); err != nil {
			return err
		}
	}
	return nil
}

func processStatusFilter(f *loop_span.FilterField) error {
	if f.QueryType == nil || *f.QueryType != loop_span.QueryTypeEnumIn {
		return fmt.Errorf("status filter should use in operator")
	}
	f.FieldName = loop_span.SpanFieldStatusCode
	f.FieldType = loop_span.FieldTypeLong
	checkSuccess, checkError := false, false
	for _, val := range f.Values {
		switch val {
		case loop_span.SpanStatusSuccess:
			checkSuccess = true
		case loop_span.SpanStatusError:
			checkError = true
		default:
			return fmt.Errorf("invalid status code field value")
		}
	}
	if checkSuccess && checkError {
		f.QueryType = ptr.Of(loop_span.QueryTypeEnumAlwaysTrue)
		f.Values = nil
	} else if checkSuccess {
		f.Values = []string{"0"}
	} else if checkError {
		f.QueryType = ptr.Of(loop_span.QueryTypeEnumNotIn)
		f.Values = []string{"0"}
	} else {
		return fmt.Errorf("invalid status code query")
	}
	return nil
}

// ms -> us
func processLatencyFilter(f *loop_span.FilterField) error {
	if f.FieldType != loop_span.FieldTypeLong {
		return fmt.Errorf("latency field type should be long ")
	}
	micros := make([]string, 0)
	for _, val := range f.Values {
		integer, err := strconv.ParseInt(val, 10, 64)
		if err != nil {
			return fmt.Errorf("fail to parse long value %s, %v", val, err)
		}
		integer = timeutil.MillSec2MicroSec(integer)
		micros = append(micros, strconv.FormatInt(integer, 10))
	}
	f.Values = micros
	return nil
}

//go:generate mockgen -destination=mocks/span_processor.go -package=mocks . TraceFilterProcessorBuilder
type TraceFilterProcessorBuilder interface {
	BuildPlatformRelatedFilter(context.Context, loop_span.PlatformType) (span_filter.Filter, error)
	BuildGetTraceProcessors(context.Context, span_processor.Settings) ([]span_processor.Processor, error)
	BuildListSpansProcessors(context.Context, span_processor.Settings) ([]span_processor.Processor, error)
	BuildAdvanceInfoProcessors(context.Context, span_processor.Settings) ([]span_processor.Processor, error)
	BuildIngestTraceProcessors(context.Context, span_processor.Settings) ([]span_processor.Processor, error)
	BuildSearchTraceOApiProcessors(context.Context, span_processor.Settings) ([]span_processor.Processor, error)
	BuildListSpansOApiProcessors(context.Context, span_processor.Settings) ([]span_processor.Processor, error)
}

type TraceFilterProcessorBuilderImpl struct {
	platformFilterFactory             span_filter.PlatformFilterFactory
	getTraceProcessorFactories        []span_processor.Factory
	listSpansProcessorFactories       []span_processor.Factory
	advanceInfoProcessorFactories     []span_processor.Factory
	ingestTraceProcessorFactories     []span_processor.Factory
	searchTraceOApiProcessorFactories []span_processor.Factory
	listSpansOApiProcessorFactories   []span_processor.Factory
}

func (t *TraceFilterProcessorBuilderImpl) BuildPlatformRelatedFilter(
	ctx context.Context,
	platformType loop_span.PlatformType,
) (span_filter.Filter, error) {
	return t.platformFilterFactory.GetFilter(ctx, platformType)
}

func (t *TraceFilterProcessorBuilderImpl) BuildGetTraceProcessors(
	ctx context.Context,
	set span_processor.Settings,
) ([]span_processor.Processor, error) {
	ret := make([]span_processor.Processor, 0)
	for _, factory := range t.getTraceProcessorFactories {
		p, err := factory.CreateProcessor(ctx, set)
		if err != nil {
			return nil, err
		}
		ret = append(ret, p)
	}
	return ret, nil
}

func (t *TraceFilterProcessorBuilderImpl) BuildListSpansProcessors(
	ctx context.Context,
	set span_processor.Settings,
) ([]span_processor.Processor, error) {
	ret := make([]span_processor.Processor, 0)
	for _, factory := range t.listSpansProcessorFactories {
		p, err := factory.CreateProcessor(ctx, set)
		if err != nil {
			return nil, err
		}
		ret = append(ret, p)
	}
	return ret, nil
}

func (t *TraceFilterProcessorBuilderImpl) BuildAdvanceInfoProcessors(
	ctx context.Context,
	set span_processor.Settings,
) ([]span_processor.Processor, error) {
	ret := make([]span_processor.Processor, 0)
	for _, factory := range t.advanceInfoProcessorFactories {
		p, err := factory.CreateProcessor(ctx, set)
		if err != nil {
			return nil, err
		}
		ret = append(ret, p)
	}
	return ret, nil
}

func (t *TraceFilterProcessorBuilderImpl) BuildIngestTraceProcessors(
	ctx context.Context,
	set span_processor.Settings,
) ([]span_processor.Processor, error) {
	ret := make([]span_processor.Processor, 0)
	for _, factory := range t.ingestTraceProcessorFactories {
		p, err := factory.CreateProcessor(ctx, set)
		if err != nil {
			return nil, err
		}
		ret = append(ret, p)
	}
	return ret, nil
}

func (t *TraceFilterProcessorBuilderImpl) BuildSearchTraceOApiProcessors(
	ctx context.Context,
	set span_processor.Settings,
) ([]span_processor.Processor, error) {
	ret := make([]span_processor.Processor, 0)
	for _, factory := range t.searchTraceOApiProcessorFactories {
		p, err := factory.CreateProcessor(ctx, set)
		if err != nil {
			return nil, err
		}
		ret = append(ret, p)
	}
	return ret, nil
}

func (t *TraceFilterProcessorBuilderImpl) BuildListSpansOApiProcessors(
	ctx context.Context,
	set span_processor.Settings,
) ([]span_processor.Processor, error) {
	ret := make([]span_processor.Processor, 0)
	for _, factory := range t.listSpansOApiProcessorFactories {
		p, err := factory.CreateProcessor(ctx, set)
		if err != nil {
			return nil, err
		}
		ret = append(ret, p)
	}
	return ret, nil
}

func NewTraceFilterProcessorBuilder(
	platformFilterFactory span_filter.PlatformFilterFactory,
	getTraceProcessorFactories []span_processor.Factory,
	listSpansProcessorFactories []span_processor.Factory,
	advanceInfoProcessorFactories []span_processor.Factory,
	ingestTraceProcessorFactories []span_processor.Factory,
	searchTraceOApiProcessorFactories []span_processor.Factory,
	listSpansOApiProcessorFactories []span_processor.Factory,
) TraceFilterProcessorBuilder {
	return &TraceFilterProcessorBuilderImpl{
		platformFilterFactory:             platformFilterFactory,
		getTraceProcessorFactories:        getTraceProcessorFactories,
		listSpansProcessorFactories:       listSpansProcessorFactories,
		advanceInfoProcessorFactories:     advanceInfoProcessorFactories,
		ingestTraceProcessorFactories:     ingestTraceProcessorFactories,
		searchTraceOApiProcessorFactories: searchTraceOApiProcessorFactories,
		listSpansOApiProcessorFactories:   listSpansOApiProcessorFactories,
	}
}
